/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package ex.learnflink;

import ex.datatypes.TaxiFare;
import ex.util.generator.TaxiFareGenerator;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 练习的任务是首先计算每个司机收集的总小费， 按小时计算，然后从中找出每小时最高的小费总额
 */
public class HourlyTipsSolution {

    private final SourceFunction<TaxiFare> source;
    private final SinkFunction<Tuple3<Long, Long, Float>> sink;

    /** Creates a job using the source and sink provided. */
    public HourlyTipsSolution(
            SourceFunction<TaxiFare> source, SinkFunction<Tuple3<Long, Long, Float>> sink) {

        this.source = source;
        this.sink = sink;
    }

    /**
     * Main method.
     *
     * @throws Exception which occurs during job execution.
     */
    public static void main(String[] args) throws Exception {
        HourlyTipsSolution job =
                new HourlyTipsSolution(new TaxiFareGenerator(), new PrintSinkFunction<>());
        job.execute();
    }

    /**
     * Create and execute the hourly tips pipeline.
     *
     * @return {JobExecutionResult}
     * @throws Exception which occurs during job execution.
     */
    public JobExecutionResult execute() throws Exception {

        // set up streaming execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //启动数据生成器并安排水印

        DataStream<TaxiFare> fares =
                env.addSource(source)
                        //生成水位线
                        .assignTimestampsAndWatermarks(
                                // 出租车的票价是合理的

                                WatermarkStrategy.<TaxiFare>forMonotonousTimestamps()

                                        .withTimestampAssigner(
                                                (fare, t) -> fare.getEventTimeMillis()));

        //计算每个司机每小时的小费
        DataStream<Tuple3<Long, Long, Float>> hourlyTips =
                fares.keyBy((TaxiFare fare) -> fare.driverId)
                        .window(TumblingEventTimeWindows.of(Time.hours(1)))//滚动时间窗口
                        .process(new AddTips());
        // 找到每小时小费总额最高的司机
        //如果不使用键控事件流，我们的程序就不能 并行 处理。
        DataStream<Tuple3<Long, Long, Float>> hourlyMax =
                hourlyTips.windowAll(TumblingEventTimeWindows.of(Time.hours(1))).maxBy(2);
        /* You should explore how this alternative (commented out below) behaves.
         * In what ways is the same as, and different from, the solution above (using a windowAll)?
         */

        // DataStream<Tuple3<Long, Long, Float>> hourlyMax = hourlyTips.keyBy(t -> t.f0).maxBy(2);

        hourlyMax.addSink(sink);

        // execute the transformation pipeline
        return env.execute("Hourly Tips");
    }

//1,2013000185,2013000185,2020-01-01T12:00:20Z,33.0
//...
//103,2013000089,2013000089,2020-01-01T12:34:20Z,41.0
//...
//128,2013000089,2013000089,2020-01-01T12:42:40Z,35.0
//...
//193,2013000150,2013000150,2020-01-01T13:04:20Z,35.0
//1> (1577883600000,2013000089,76.0)
//194,2013000073,2013000073,2020-01-01T13:04:40Z,2.0
//...
//223,2013000197,2013000197,2020-01-01T13:14:20Z,35.0
//...
//238,2013000197,2013000197,2020-01-01T13:19:20Z,36.0
//...
//373,2013000190,2013000190,2020-01-01T14:04:20Z,22.0
//2> (1577887200000,2013000197,71.0)
//374,2013000114,2013000114,2020-01-01T14:04:40Z,23.0
//...

    /*
     * Wraps the pre-aggregated result into a tuple along with the window's timestamp and key.
     */
    public static class AddTips
            extends ProcessWindowFunction<TaxiFare, Tuple3<Long, Long, Float>, Long, TimeWindow> {


        @Override
        public void process(Long key, Context context, Iterable<TaxiFare> fares, Collector<Tuple3<Long, Long, Float>> out) throws Exception {
            float sumOfTips = 0F;
            for (TaxiFare f : fares) {
                sumOfTips += f.tip;
            }
            out.collect(Tuple3.of(context.window().getEnd(), key, sumOfTips));
        }
    }
}
